package com.wfg.flink.core.example;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.time.Instant;

public class FlinkWatermarkDemo {

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 假设我们有一个数据源，这里使用 fromElements 模拟
        DataStream<Event> eventStream = env.fromElements(
                new Event(1L, Instant.now().minusSeconds(10).toEpochMilli()),
                new Event(2L, Instant.now().minusSeconds(8).toEpochMilli()),
                new Event(3L, Instant.now().minusSeconds(12).toEpochMilli()),
                new Event(4L, Instant.now().minusSeconds(15).toEpochMilli()),
                new Event(5L, Instant.now().minusSeconds(19).toEpochMilli()),
                new Event(6L, Instant.now().minusSeconds(18).toEpochMilli()),
                new Event(7L, Instant.now().minusSeconds(22).toEpochMilli())
        );

        // 定义 Watermark 策略，允许 5 秒的乱序
        WatermarkStrategy<Event> watermarkStrategy = WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> {
                    // 从事件中提取时间戳
                    return event.getTimestamp();
//                    timestamp.assignTimestamp(event.getTimestamp());
                });

        // 应用 Watermark 策略，并处理数据流
        DataStream<String> resultStream = eventStream
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .keyBy(event -> event.getId()) // 根据事件 ID 进行分区（这只是一个示例，实际可能根据业务需求分区）
                // 接下来可以进行窗口操作、时间聚合等操作
                .map(new MapFunction<Event, String>() {
                    @Override
                    public String map(Event event) throws Exception {
                        return "Event ID: " + event.getId() + ", Timestamp: " + Instant.ofEpochMilli(event.getTimestamp());
                    }
                });

        // 输出结果
        resultStream.print();

        // 执行任务
        env.execute("Flink Watermark Demo");
    }

    // 定义一个简单的事件类
    public static class Event {
        private final long id;
        private final long timestamp;

        public Event(long id, long timestamp) {
            this.id = id;
            this.timestamp = timestamp;
        }

        public long getId() {
            return id;
        }

        public long getTimestamp() {
            return timestamp;
        }
    }
}